package org.geektimes.reactive.message.streams;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.LinkedList;
import java.util.List;

/**
 * @Author: jipeng
 * @Description:
 * @Date: Created in 2021/4/6 8:27
 */
public class DefaultPublisher<T> implements Publisher<T> {

    public List<Subscriber> subscribers = new LinkedList<>();


    @Override
    public void subscribe(Subscriber<? super T> s) {
        DefaultSubscription subscription = new DefaultSubscription(s);
        s.onSubscribe(subscription);

        subscribers.add(new SubscriberWrapper(s, subscription));
    }


    public void publish(T data) {
        //广播
        subscribers.forEach(subscriber -> {
            SubscriberWrapper subscriberWrapper = SubscriberWrapper.class.cast(subscriber);

            // 判断当前 subscriber 是否 cancel  数据发送
            if (subscriberWrapper.getSubscription().isCanceled()) {
                System.err.println("本次数据发布已忽略，数据为 ：" + data);
                return;
            }
            //继续发送
            subscriber.onNext(data);
        });
    }


    public void error(Throwable error) {
        //广播
        subscribers.forEach(subscriber -> {
            subscriber.onError(error);
        });
    }


    public void onComplete() {
        //广播
        subscribers.forEach(subscriber -> {
            subscriber.onComplete();
        });
    }


    public static void main(String[] args) {
        DefaultPublisher publisher = new DefaultPublisher();
        publisher.subscribe(new DefaultSubscriber());
        for (int i = 0; i < 5; i++) {
            publisher.publish(i);
        }

    }
}
